package com.offcn.spark.p3

import java.sql.DriverManager

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

/**
 * @Auther: BigData-LGW
 * @ClassName: ForeachPartitionOps
 * @Date: 2020/12/7 23:26
 * @功能描述: $FunctionDescription
 * @Version:1.0
 */
object ForeachPartitionOps {
    def main(args: Array[String]): Unit = {
        val conf = new SparkConf()
            .setAppName("_05ActionOps")
            .setMaster("local[*]")
        val sc = new SparkContext(conf)
        val array = sc.parallelize(Array(
            "hello you",
            "hello me",
            "hello you",
            "hello you",
            "hello me",
            "hello you"
        ), 2)
        val pairs = array.flatMap(_.split("\\s+")).map((_,1))
        val ret = pairs.aggregateByKey(0)(_+_,_+_)
//        saveInfoMySQL(ret)
        saveInfoMySQL2(ret)
        sc.stop()
    }
    def saveInfoMySQL2(rdd:RDD[(String,Int)])={
        rdd.foreach{
            case (word,count) => {
                Class.forName("com.mysql.jdbc.Driver")
                val url = "jdbc:mysql://localhost:3306/wordcount"
                val connection = DriverManager.getConnection(url,"mark","sorry")
                val sql =
                    """
                      |insert into wordcounts(word,`count`) value(?,?)
                      |""".stripMargin
                val ps = connection.prepareStatement(sql)
                ps.setString(1,word)
                ps.setInt(2,count)
                ps.execute()
                ps.close()
                connection.close()
            }
        }
    }
    def saveInfoMySQL(rdd:RDD[(String,Int)])={
        Class.forName("com.mysql.jdbc.Driver")
        val url = "jdbc:mysql://localhost:3306/wordcount"
        val connection = DriverManager.getConnection(url,"mark","sorry")
        val sql =
            """
              |insert into wordcounts(word,`count`) value(?,?)
              |""".stripMargin
        val ps = connection.prepareStatement(sql)
        rdd.foreach{
            case (word,count) => {
                ps.setString(1,word)
                ps.setInt(2,count)
                ps.execute()
            }
        }
        ps.close()
        connection.close()
    }
}
